package org.geektimes.reactive.streams;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * 业务消息 订阅者
 */                                                                       //订阅者
public class BusinessSubscriber<T> implements Subscriber<T> {

    private Subscription subscription;//订阅（名词）

    private int count = 0;

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
    }

    @Override
    public void onNext(T t) {
        System.out.println("收到数据："+t);
        count++;
        if(count>2){// 当到达数据阈值时，取消 Publisher 给当前 Subscriber 发送数据
            subscription.cancel();
        }
    }

    @Override
    public void onError(Throwable t) {
        System.out.println("异常："+t);
    }

    @Override
    public void onComplete() {
        System.out.println("收到数据完成");
    }
}
